/*
    Copyright 2010 Northbranchlogic, Inc.

    Licensed under the Apache License, Version 2.0 (the "License");
    you may not use this file except in compliance with the License.
    You may obtain a copy of the License at

        http://www.apache.org/licenses/LICENSE-2.0

    Unless required by applicable law or agreed to in writing, software
    distributed under the License is distributed on an "AS IS" BASIS,
    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    See the License for the specific language governing permissions and
    limitations under the License.
 */

package com.norbl.runjrun;

import com.norbl.util.*;
import com.norbl.util.ec2.*;
import com.norbl.util.sdb.*;
import com.xerox.amazonws.sdb.*;
import com.xerox.amazonws.ec2.*;
import java.util.*;

/** Provides access to the SDB 'table' of {@link NodeState}s,
 *  where each element represents one instance of {@link RunnerNode}.<p> 
 *
 * @author Barnet Wagman
 */

public class NodeStateTable extends DomainWrangler {

    Ec2Ops eops;

    private long runnerNodePulseIntervalMillis;

    public NodeState[] currentStates;

    public NodeStateTable(SdbOps dops, Ec2Ops eops,
                          long runnerNodePulseIntervalMillis)
        throws SDBException {

        super(dops,Constants.NODE_TABLE_NAME);
        this.eops = eops;
        this.runnerNodePulseIntervalMillis = runnerNodePulseIntervalMillis;
    }

    public NodeStateTable cleanClone(SdbOps newDops,Ec2Ops newEops) {
        try {
            return(new NodeStateTable(newDops,newEops,
                                      runnerNodePulseIntervalMillis) );
        }
        catch(SDBException x) { throw new RuntimeException(x); }
    }

    public DbEl toDbEl(PropertyList pl,boolean allowMissingFields) {
        return(new NodeState(pl,true));
    }

    public DbEl getTemplateDbEl() { return(new NodeState()); }

    public NodeState[] getNodeStates() throws SDBException {
        List<DbEl> els = getItems(true);        
        return( els.toArray(new NodeState [els.size()]) );
    }

        // ------ State setters --------------------------------------

    public void setState(String hostName,NodeState nodeState, boolean retry)
        throws SDBException {
        putItem(hostName,nodeState,retry);     
    }
    
    public void setInstanceState(String hostName,
                                 String instanceID,
                                 NodeState.InstanceState instanceState,
                                 Long statusTime,
                                 boolean retry)
        throws SDBException {

        putItem(hostName,
                new NodeState(hostName,
                              instanceID,
                              instanceState,
                              null, // launch time
                              null, // NodeState.Status.idle,
                              statusTime,
                              null // running task
                              ),
                retry);

    }

    public void setInstanceStateNil(String hostName,
                                    Long statusTime,
                                    boolean retry)
        throws SDBException {

        putItem(hostName,
                new NodeState(Constants.NIL,
                              Constants.NIL,
                              NodeState.InstanceState.nil,
                              null, // launch time
                              NodeState.Status.nil, // NodeState.Status.idle,
                              statusTime,
                              null // running task
                              ),
                retry);

    }

    public void setNodeStateNil(String hostName,
                                    Long statusTime,
                                    boolean retry)
        throws SDBException {

        putItem(hostName,
                new NodeState(null, // host
                              null, // instance
                              null, // instance state
                              null, // launch time
                              NodeState.Status.nil, // NodeState.Status.idle,
                              statusTime,
                              null // running task
                              ),
                retry);

    }

        // --------------------------------------------------------

        /** 
         *
         * @throws SDBException
         * @throws EC2Exception
         * @return true if any node states were changed, else false.
         */
    public boolean update() throws SDBException, EC2Exception {

        long curTm = System.currentTimeMillis();

            // Get all of user's instances in a ht keyed by instance ID
        HashMap<String,String> insStateHt = eops.getInstanceStates();
       
            // Get all nodes
        currentStates = getNodeStates();
        
            // Set the instance state for each node
        boolean anyChanged = false;
        for (NodeState node : currentStates ) {
            
            if ( checkNodePulse(node,curTm) ) {              
                anyChanged = true;
            }

            if ( node.isOnEc2Instance() ) {
                if ( checkInstanceState(node,insStateHt) ) {
                    setState(node.host,node,true);
                    anyChanged = true;
                }
            }
            else { // The node is on an non-ec2 instance system.               
                if ( !NodeState.InstanceState.not_ec2.equals(node.instanceState) ) {
                    node.instanceState = NodeState.InstanceState.not_ec2;
                    setState(node.host,node,true);
                    anyChanged = true;
                }
            }            
        }
        return(anyChanged);
    }
    
        /** Deletes all nodes in the node state table. Note that this
         *  does not stop the node app or kill instances. This method
         *  blocks until the table is actually empty.
         */
    public void clear() throws SDBException {
        
        deleteAllItems(true);

        int n = Integer.MAX_VALUE;
        while ( (n = getItems(true).size()) > 0 ) {
            Time.nap(250L);
        }
    }

        /** Delete any node whose ec2 instance is not in state 'running'.
         *
         */
    public int clearDeadInstances() throws SDBException {

        NodeState[] nodes = getNodeStates();

        int nDeleted = 0;
        for ( NodeState s : nodes ) {
            if ( !s.instanceState.equals(NodeState.InstanceState.running) ) {
                deleteItem(s.host,true);
                ++nDeleted;
            }
        }
        return(nDeleted);
    }

    public void delete(String instanceID) throws SDBException {
        dops.deleteItems(Constants.NODE_TABLE_NAME,
                         "select * from " + Constants.NODE_TABLE_NAME +
                         " where instanceID = " +
                         "\"" + instanceID + "\"", // query string
                         true);
    }

        // ---------------------------------------------

    public static NodeState getNode(NodeState[] nodes,String host) {

        if ( host == null ) return(null);

        for ( NodeState s : nodes ) {
            if ( host.equals(s.host) ) return(s);
        }

        return(null);
    }

     public NodeState getNode(String host) {

        if ( host == null ) return(null);

        for ( NodeState s : currentStates ) {
            if ( host.equals(s.host) ) return(s);
        }

        return(null);
    }

        // ----------------------------------------------

        /** If too much time has elapsed since the last pulse, we declare
         *  the node 'dead' and set it's state to no_pulse.
         * 
         * @param s
         * @param tm
         * @return true if the status has been changed, else false.
         */
    private boolean checkNodePulse(NodeState s, long tm) {

        if ( !s.instanceIsRunning() ) return(false);
        if ( s.status.equals(NodeState.Status.nil) ) return(false);

        long dt = tm - s.statusTime.longValue();
        if ( (dt > (2*runnerNodePulseIntervalMillis)) &&
             !s.status.equals(NodeState.Status.no_pulse) ) {
                // State has changed to no pulse.
            s.status = NodeState.Status.no_pulse;
            return(true);
        }
        else return(false);
    }

        /** If the instance state as changed per the ins ht, this
         *  methods update the instance state in the NodeState object s.
         *
         * @param s
         * @param insStateHt
         * @return
         */
    private boolean checkInstanceState(NodeState s,
                                       HashMap<String,String> insStateHt) {

        if ( !s.isOnEc2Instance() ) return(false);

        NodeState.InstanceState insState =
            NodeState.InstanceState.stateOf(insStateHt.get(s.instanceID));

        boolean changed = false;

        if ( !insState.equals(NodeState.InstanceState.running) ) {
                // If the instance is not running, then the node must be NIL
            if ( !s.status.equals(NodeState.Status.nil) ) {               
                s.status = NodeState.Status.nil;
                s.instanceState = insState;
                changed = true;               
            }
        }

        if ( !insState.equals(s.instanceState) ) { // The instance state HAS changed.
            changed = true;             
            s.instanceState = insState;           
        }

        return(changed);
    }
}
